package cn.sh.library.mgts.framework.factory;

import java.io.IOException;

import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.sh.library.mgts.common.Constant;

public class ESClientSpringFactory {
	public static int CONNECT_TIMEOUT_MILLIS = 1000;
    public static int SOCKET_TIMEOUT_MILLIS = 30000;
    public static int CONNECTION_REQUEST_TIMEOUT_MILLIS = 500;
    public static int MAX_CONN_PER_ROUTE = 10;
    public static int MAX_CONN_TOTAL = 30;
    public static String USER = "";
    public static String PASSWORD = "";

    private static HttpHost[] HTTP_HOSTS;
    private RestClientBuilder builder;
    private RestClient restClient;
    private RestHighLevelClient restHighLevelClient;
    private BulkProcessor.Builder bulkProcessorBuilder;

    private static ESClientSpringFactory esClientSpringFactory = new ESClientSpringFactory();
	private static Logger logger = LoggerFactory.getLogger(ESClientSpringFactory.class);
	private static Logger loggerBulk = LoggerFactory.getLogger(Constant.LOGGER_BULK_NAME);

    private ESClientSpringFactory(){}

    public static ESClientSpringFactory build(HttpHost[] httpHosts,
                                              Integer maxConnectNum, Integer maxConnectPerRoute){
        HTTP_HOSTS = httpHosts;
        MAX_CONN_TOTAL = maxConnectNum;
        MAX_CONN_PER_ROUTE = maxConnectPerRoute;
        return  esClientSpringFactory;
    }
    

    public static ESClientSpringFactory build(HttpHost[] httpHosts,
                                              Integer maxConnectNum, Integer maxConnectPerRoute, String user, String password){
        USER = user;
        PASSWORD = password;
    	HTTP_HOSTS = httpHosts;
        MAX_CONN_TOTAL = maxConnectNum;
        MAX_CONN_PER_ROUTE = maxConnectPerRoute;
        return  esClientSpringFactory;
    }

    public static ESClientSpringFactory build(HttpHost[] httpHosts,Integer connectTimeOut, Integer socketTimeOut,
                                              Integer connectionRequestTime,Integer maxConnectNum, Integer maxConnectPerRoute){
    	HTTP_HOSTS = httpHosts;
        CONNECT_TIMEOUT_MILLIS = connectTimeOut;
        SOCKET_TIMEOUT_MILLIS = socketTimeOut;
        CONNECTION_REQUEST_TIMEOUT_MILLIS = connectionRequestTime;
        MAX_CONN_TOTAL = maxConnectNum;
        MAX_CONN_PER_ROUTE = maxConnectPerRoute;
        return  esClientSpringFactory;
    }

    public void init(){
        builder = RestClient.builder(HTTP_HOSTS);
        setConnectTimeOutConfig();
        setMutiConnectConfig();
        restClient = builder.build();
        restHighLevelClient = new RestHighLevelClient(builder);
        initBulkProcess();
    }
    // 配置连接时间延时
    public void setConnectTimeOutConfig(){
        builder.setRequestConfigCallback(requestConfigBuilder -> {
            requestConfigBuilder.setConnectTimeout(CONNECT_TIMEOUT_MILLIS);
            requestConfigBuilder.setSocketTimeout(SOCKET_TIMEOUT_MILLIS);
            requestConfigBuilder.setConnectionRequestTimeout(CONNECTION_REQUEST_TIMEOUT_MILLIS);
            return requestConfigBuilder;
        });
    }
    
    // 使用异步httpclient时设置并发连接数
    public void setMutiConnectConfig(){
        final CredentialsProvider credentialsProvider =
        	    new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
        	    new UsernamePasswordCredentials(USER, PASSWORD));
            
        builder.setHttpClientConfigCallback(httpClientBuilder -> {
            httpClientBuilder.setMaxConnTotal(MAX_CONN_TOTAL);
            httpClientBuilder.setMaxConnPerRoute(MAX_CONN_PER_ROUTE);
        	if(!StringUtils.isEmpty(USER)) {
        		httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
        	}
            return httpClientBuilder;
        });
    }

    public RestClient getClient(){
        return restClient;
    }

    public RestHighLevelClient getRhlClient(){
        return restHighLevelClient;
    }
    
    public void initBulkProcess() {
    	BulkProcessor.Listener listener = new BulkProcessor.Listener() {
		    @Override
		    public void beforeBulk(long executionId, BulkRequest request) {
//		        int numberOfActions = request.numberOfActions(); 
//		        logger.debug("Executing bulk [{}] with {} requests",
//		                executionId, numberOfActions);
		    }

		    @Override
		    public void afterBulk(long executionId, BulkRequest request,
		            BulkResponse response) {
		        if (response.hasFailures()) { 
		        	loggerBulk.warn("Bulk [{}] executed with failures", executionId);
		        	loggerBulk.error(response.buildFailureMessage());
		        } else {
		        	loggerBulk.debug("Bulk [{}] completed in {} milliseconds",
		                    executionId, response.getTook().getMillis());
		        }
		    }

		    @Override
		    public void afterBulk(long executionId, BulkRequest request,
		            Throwable failure) {
		    	loggerBulk.error("Failed to execute bulk", failure); 
		    }
		};
		

		bulkProcessorBuilder = BulkProcessor.builder(
		        (request, bulkListener) ->
		        restHighLevelClient.bulkAsync(request, RequestOptions.DEFAULT, bulkListener),
		        listener);
		bulkProcessorBuilder.setBulkActions(500); 
		bulkProcessorBuilder.setBulkSize(new ByteSizeValue(5L, ByteSizeUnit.MB)); 
		bulkProcessorBuilder.setConcurrentRequests(10); 
		bulkProcessorBuilder.setFlushInterval(TimeValue.timeValueSeconds(10L)); 
		bulkProcessorBuilder.setBackoffPolicy(BackoffPolicy
		        .constantBackoff(TimeValue.timeValueSeconds(1L), 3));
    }
    
    public BulkProcessor.Builder getBulkProcessorBuilder(){
    	return bulkProcessorBuilder;
    }
    
    public BulkProcessor getBulkProcessor(){
    	return bulkProcessorBuilder.build();
    }
    
    public static BulkProcessor createBulkProcessorBuilder(RestHighLevelClient restHighLevelClient) {

    	BulkProcessor.Listener listener = new BulkProcessor.Listener() {
		    @Override
		    public void beforeBulk(long executionId, BulkRequest request) {
//		        int numberOfActions = request.numberOfActions(); 
//		        logger.debug("Executing bulk [{}] with {} requests",
//		                executionId, numberOfActions);
		    }

		    @Override
		    public void afterBulk(long executionId, BulkRequest request,
		            BulkResponse response) {
		        if (response.hasFailures()) { 
		        	loggerBulk.warn("Bulk [{}] executed with failures", executionId);
		        	loggerBulk.error(response.buildFailureMessage());
		        } else {
		        	loggerBulk.debug("Bulk [{}] completed in {} milliseconds",
		                    executionId, response.getTook().getMillis());
		        }
		    }

		    @Override
		    public void afterBulk(long executionId, BulkRequest request,
		            Throwable failure) {
		    	loggerBulk.error("Failed to execute bulk", failure); 
		    }
		};
		

		BulkProcessor.Builder bulkProcessorBuilder = BulkProcessor.builder(
		        (request, bulkListener) ->
		        restHighLevelClient.bulkAsync(request, RequestOptions.DEFAULT, bulkListener),
		        listener);
		bulkProcessorBuilder.setBulkActions(500); 
		bulkProcessorBuilder.setBulkSize(new ByteSizeValue(5L, ByteSizeUnit.MB)); 
		bulkProcessorBuilder.setConcurrentRequests(10); 
		bulkProcessorBuilder.setFlushInterval(TimeValue.timeValueSeconds(10L)); 
		bulkProcessorBuilder.setBackoffPolicy(BackoffPolicy
		        .constantBackoff(TimeValue.timeValueSeconds(1L), 3));
    	return bulkProcessorBuilder.build();
    }
    

    public void close() {
        if (restClient != null) {
            try {
                restClient.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    
}
